package com.spark.core.examples

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.util.control.Breaks._

/**
 * 分组取topN问题
 * 定义定长数组
 */
object TopN2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("topN")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val infos = sc.textFile("./data/scores.txt")
    val pairInfo: RDD[(String, Int)] = infos.map(one => {
      (one.split("\t")(0), one.split("\t")(1).toInt)
    })

    val result: Array[(String, mutable.Buffer[Int])] = pairInfo.groupByKey().map(tup => {
      val grade: String = tup._1
      val iter: Iterator[Int] = tup._2.iterator

      // 动态数组
      val top3Score = new Array[Int](3)

      while (iter.hasNext) {
        val currScore: Int = iter.next()
        breakable {
          for (i <- 0 until top3Score.length) {
            if (top3Score(i) == 0) {
              top3Score(i) = currScore
              break
            } else if (currScore > top3Score(i)) {
              for (j <- top3Score.length - 1 until(i, -1)) {
                top3Score(j) = top3Score(j - 1)
              }
              top3Score(i) = currScore
              break()
            }
          }
        }
      }
      (grade, top3Score.toBuffer)
    }).collect()

    result.foreach(println)

  }
}
